//package vip.huhailong.redismq.redistool;

//
//import com.alibaba.fastjson.JSONObject;
//import com.sun.xml.internal.ws.message.stream.StreamMessage;
//import org.apache.commons.lang3.StringUtils;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.data.redis.connection.stream.ObjectRecord;
//import org.springframework.data.redis.connection.stream.RecordId;
//import org.springframework.data.redis.stream.StreamListener;
//import org.springframework.stereotype.Component;
//
//import java.lang.invoke.MethodHandles;
//import java.util.concurrent.TimeUnit;
//
///**
// * @ClassName: ListenerMessage
// * @date: 30/5/2022 上午11:00
// * @Description: 监听圈人消息处理类
// */
//@Component
//public class MessageListener implements StreamListener<String, ObjectRecord<String, StreamMessage>> {
//
//    protected static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
//    @Autowired
//    private RedisStream redisStream;
//    @Value("${spring.redis.stream.group}")
//    private String group;//消费组
//    @Value("${message.repeatTimes:2}")
//    private String repeatTimes;//重试最大次数
//    @Autowired
//    private IThirdpartyService thirdpartyService;
//
//    @Override
//    public void onMessage(ObjectRecord<String, StreamMessage> message) {
//        // 接收到消息
//        LOGGER.info("==圈人引擎接收到圈人消息，内容：{}",JSONObject.toJSONString(message));
//        RecordId recordId = message.getId();//消息记录对象
//        String recordVal = recordId.getValue();//消息记录id
//        String stream = message.getStream();//消息stream name
//        StreamMessage body = message.getValue();//消息内容对象
//        LOGGER.info("==消息id: {}, stream: {}, body: {}", recordId, stream, JSONObject.toJSONString(body));
//        if(body != null){
//            LogUtils.setLogTrace(body);//增加日志traceId
//            long taskId = body.getTaskId();//任务id
//            String segmentId = String.valueOf(body.getBusinessId());//人群id
//            //查询圈人配置
//            SegmentTaskResDTO segmentTaskResDTO = thirdpartyService.getTaskConfig(taskId,segmentId);
//            if(segmentTaskResDTO != null){
//                String engineType = segmentTaskResDTO.getEngineType();//圈人类型
//                String lockKey = SegmentConstants.ENGINE_LOCK_KEY + segmentId;//redis锁key
//                String lockVal = "lock_"+segmentId;//redis锁val
//                try{
//                    //调用web端接口更新segment_task_history表状态为处理中
//                    thirdpartyService.updateTaskStatus(taskId,segmentId,TaskStatusEnum.EXECUTE.getCode());
//
//                    //处理业务逻辑
//                    ISegmentService service = ServiceFactory.getService(engineType);
//                    if(StringUtils.isNotBlank(segmentId) && service != null){
//                        //添加分布锁，防止多节点重复执行
//                        Boolean success = redisStream.lock(lockKey,lockVal,10,TimeUnit.MINUTES);
//                        if(success){
//                            service.segmentHandle(taskId,segmentId,segmentTaskResDTO.getContent());
//                        }else{
//                            LOGGER.info("获取分布式锁失败！存在其他节点执行该圈人任务，lockKey={}",lockKey);
//                        }
//                    }
//
//                    this.unlock(lockKey,lockVal);
//                }catch (Exception e){
//                    LOGGER.error("圈人消费者处理消息异常！",e);
//                    this.unlock(lockKey,lockVal);
//
//                    //查询redis 重试次数是否小于2，是的话，将消息放入队列进行重试
//                    String repeatKey = SegmentConstants.MESSAGE_REPEAT_TIMES_KEY+taskId;
//                    try {
//                        Integer times = (Integer) redisStream.get(repeatKey);
//                        if(times == null || times.intValue() < Integer.valueOf(repeatTimes)){
//                            //消息放入队列，重试
//                            thirdpartyService.sendMessage(taskId, segmentId, body.getRedisStream());
//
//                            Integer cacheTimes = times == null? 1: times +1;
//                            if(times == null){
//                                redisStream.set(repeatKey,cacheTimes,3600*60*24);//24小时后过期
//                            }else{
//                                redisStream.set(repeatKey,cacheTimes);
//                            }
//
//                            //调用web接口修改重试次数
//                            thirdpartyService.updateRepeatNumber(taskId, segmentId);
//                        }else{
//                            //调用web端接口更新segment_task_history表状态为失败
//                            thirdpartyService.updateTaskStatus(taskId,segmentId,TaskStatusEnum.EXECUTE_FAIL.getCode());
//                        }
//                    } catch (Exception ex) {
//                        LOGGER.error("获取redis 重试次数异常！taskId="+taskId,ex);
//                    }
//                }finally {
//                    // 消费完成后确认消费（ACK）
//                    this.ackMessage(stream,recordVal);
//                }
//            }else{
//                LOGGER.info("查询到的圈人配置内容为null, taskId={}, segmentId={}",taskId,segmentId);
//                this.ackMessage(stream,recordVal);
//                //设置任务状态为圈人失败
//                thirdpartyService.updateTaskStatus(taskId,segmentId,TaskStatusEnum.EXECUTE_FAIL.getCode());
//            }
//        }else {
//            LOGGER.info("==消息id: {}, stream: {}, 消息内容body为空，无法进行圈人操作！", recordId, stream);
//            this.ackMessage(stream,recordVal);
//        }
//    }
//
//    private void unlock(String lockKey,String lockVal){
//        //释放锁
//        try{
//            redisStream.unlock(lockKey,lockVal,0, TimeUnit.MINUTES);
//            LOGGER.info("释放分布式锁成功！lockKey={}",lockKey);
//        }catch (Exception ex){
//            LOGGER.error("释放分布式锁异常！lockKey={}",lockKey);
//        }
//    }
//
//    private void ackMessage(String stream,String recordVal){
//        Long num = redisStream.ack(stream,group, recordVal);
//        if(num != null && num > 0){
//            LOGGER.info("圈人消息确认成功！stream={}, group={}, recordId={}",stream,group,recordVal);
//        }
//        //删除已确认消息
//        redisStream.deleteRecord(stream,recordVal);
//        LOGGER.info("圈人消息删除成功！stream={}, group={}, recordId={}",stream,group,recordVal);
//    }
//
//}
